package org.netty.framework.websocket.handle;

import java.lang.reflect.Method;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.netty.framework.core.annotation.HandlerExecutionChain;
import org.netty.framework.core.util.AnnotationContainer;
import org.netty.framework.websocket.websocket.WebSocketRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelId;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpUtil;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.util.concurrent.GlobalEventExecutor;

/**
 * 处理websocket请求
 * 
 * @author chenfanglin
 * @date 2018年10月9日下午7:27:15
 *
 */
public class WebSocketChannelHandle extends SimpleChannelInboundHandler<Object> {

	private static final Logger logger = LoggerFactory.getLogger(WebSocketChannelHandle.class);

	public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

	public static Map<ChannelId, HandlerExecutionChain> channelMap = new ConcurrentHashMap<>();
	// private NettyServerThreadPoolExecutor nettyServerThreadPoolExecutor;
	//
	// public WebSocketChannelHandle() {
	// nettyServerThreadPoolExecutor =
	// WebSocketServerThreadPoolExecutor.getInstance().getThreadPoolExecutor();
	// }

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		Channel newChanel = ctx.channel();
		for (Channel channel : channels) {
			channel.writeAndFlush(new TextWebSocketFrame("建立一个新连接:" + newChanel.id().asLongText()));
		}
		channels.add(ctx.channel());
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		logger.info("Channel发生异常:{},channelId:{}", cause, ctx.channel().id());
		ctx.close();
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
		Channel newChannel = ctx.channel();
		if (msg instanceof FullHttpRequest) {
			/**
			 * websocket 握手之前会发送一条http请求过来
			 */
			FullHttpRequest request = (FullHttpRequest) msg;
			String url = request.uri();
			HandlerExecutionChain handlerExecutionChain = AnnotationContainer.getWebSocketHandle(url);
			if (handlerExecutionChain == null) {
				sendNotFoundResponse(newChannel);
				return;
			}
			String webSocketURL = "ws://" + request.headers().get(HttpHeaderNames.HOST) + url;
			WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(webSocketURL, null,
					false);
			WebSocketServerHandshaker handshaker = wsFactory.newHandshaker(request);
			if (handshaker == null) {
				WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(newChannel);
			} else {
				// 告诉客户端,完成握手
				handshaker.handshake(ctx.channel(), request);
				channelMap.put(newChannel.id(), handlerExecutionChain);
			}
		} else if (msg instanceof WebSocketFrame) {
			/**
			 * 处理websocket的消息
			 */
			if (msg instanceof TextWebSocketFrame) {
				TextWebSocketFrame frame = (TextWebSocketFrame) msg;
				WebSocketRequest request = new WebSocketRequest(frame.text());
				HandlerExecutionChain handlerExecutionChain = channelMap.get(newChannel.id());
				Method method = handlerExecutionChain.getMethod();
				Object result = method.invoke(handlerExecutionChain.getHandle(), request);
				writeResponse(newChannel, result);
			}
		} else {
			WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(newChannel);
		}
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		Channel newChanel = ctx.channel();
		for (Channel channel : channels) {
			channel.writeAndFlush(new TextWebSocketFrame("断开一个连接:" + newChanel.id()));
		}
		channels.remove(newChanel);
	}

	private void writeResponse(Channel channel, Object result) {
		channel.writeAndFlush(new TextWebSocketFrame(String.valueOf(result)));
	}

	/**
	 * 请求路径不存在
	 * 
	 * @param channel
	 */
	private void sendNotFoundResponse(Channel channel) {
		HttpResponse res = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_FOUND);
		HttpUtil.setContentLength(res, 0);
		channel.writeAndFlush(res);
	}

}
